package org.huangrui.spark.scala.core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}


/**
 * @Author hr
 * @Create 2024-10-19 2:30
 */
object HotCategoryTop10Analysis_2 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("spark")
    val sc = new SparkContext(conf)
    // TODO 将文件作为数据源，对接RDD进行操作
    val data = sc.textFile("data/user_visit_action.txt")
    // Q : 存在大量的shuffle操作（reduceByKey）
    // reduceByKey 聚合算子，spark会提供优化，缓存


    // TODO 1. 将多余的数据进行删除（过滤）处理
    val filterRdd = data.filter((line: String) => {
      val strings = line.split("_")
      "null".matches(strings(5))
    })
    // 2. 将数据转换结构
    //    点击的场合:(品类ID,(1,0,0))
    //    下单的场合:(品类ID,(0,1,0))
    //    支付的场合:(品类ID,(0,0,1))
    val flatRdd: RDD[(String, (Int, Int, Int))] = filterRdd.flatMap {
      action: String => {
        val datas = action.split("_")
        if (datas(6) != "-1") {
          // 点击的场合
          List((datas(6), (1, 0, 0)))
        } else if (datas(8) != "null") {
          // 下单的场合
          val ids = datas(8).split(",")
          ids.map((id: String) => (id, (0, 1, 0)))
        } else if (datas(10) != "null") {
          // / 支付的场合
          val ids = datas(10).split(",")
          ids.map((id: String) => (id, (0, 0, 1)))
        } else {
          Nil
        }
      }
    }
    // 3. 将相同的品类ID的数据进行分组聚合
    //    (品类ID,(点击数量, 下单数量, 支付数量))
    flatRdd.reduceByKey((x: (Int, Int, Int), y: (Int, Int, Int)) => {
      Tuple3(x._1 + y._1, x._2 + y._2, x._3 + y._3)
    }).sortBy(_._2, ascending = false).take(10).foreach(println)

    sc.stop()
  }
}
